package DB;

import java.util.Vector;

public class Join extends RAQuery {

	private RAQuery raQuery2;
	private Vector<String> titles;
	private Vector<String> keys;

/*
 * Constructor
 */
	
	public Join(RAQuery raq1, RAQuery raq2) {
		this.raQuery  = raq1;
		this.raQuery2 = raq2;
		titles = new Vector<String>();
		keys   = new Vector<String>();
	}

/*
 * @see DB.RAQuery#compute()
 * Computing join of two relational algebra queries 
 */
	
	@Override
	public Table compute() {
		Table t1 = raQuery.compute();
		Table t2 = raQuery2.compute();
		Vector<String> common = buildTitlesAndKeys(t1, t2);
		Table ans = new Table("", keys, titles, t1.getTotalProb() + t2.getTotalProb());
		Vector<String> iTuple;
		int[] comIndT1 = new int[common.size()];
		int[] comIndT2 = new int[common.size()];
		for (int i = 0 ; i < common.size(); i++){
			comIndT1[i]= t1.getTitles().indexOf(common.elementAt(i)); // if there is more then 1 common field?
			comIndT2[i]= t2.getTitles().indexOf(common.elementAt(i));
		}
		// In case the join is between two tables that have
		// a common field, the merging needs to be corresponded
		// to it (natural join)
		if (common.size()>0){
			for(int i = 0; i < t1.getSize(); i++){
				iTuple = t1.getTuple(i);
				for (int j = 0 ; j < t2.getSize() ; j++ )
					if (tuplesMatch(iTuple, t2.getTuple(j), comIndT1, comIndT2)){
						Vector<String> tupleToAdd = new Vector<String>();
						tupleToAdd.addAll(iTuple);
						addOtherFields(tupleToAdd, t2.getTuple(j), comIndT2);
						ans.addTuple(tupleToAdd, t1.getTupleProp(i) * t2.getTupleProp(j));
					}				
			}
		}
		return ans;	
	}

/*
 * Adds all the necessary fields to the tuple, when the
 * common field is added only once
 */
	private void addOtherFields(Vector<String> tupleToAdd, Vector<String> tuple, int[] comInd) {
		Vector<String> tTuple = new Vector<String>();
		tTuple.addAll(tuple);
		for(int i = 0; i < comInd.length; i++){
			tTuple.remove(comInd[i]);
		}
		tupleToAdd.addAll(tTuple);
	}

/*
 * Match checking if value in the common field 
 * in the two tuples
 */
	private boolean tuplesMatch(Vector<String> t1, Vector<String> t2, int[] comIndT1, int[] comIndT2) {
		for(int i = 0; i < comIndT1.length; i++){
			if(!t1.elementAt(comIndT1[i]).equalsIgnoreCase(t2.elementAt(comIndT2[i])))
				return false;
		}
		return true;
	}

/*
 * building the fields and the keys of the new table by merging 
 * between those of first Table and the second Table 
 */
	private Vector<String> buildTitlesAndKeys(Table t1, Table t2) {
		Vector<String> ans = new Vector<String>();
		Vector<String> t1Titles = t1.getTitles();
		Vector<String> t1Keys = t1.getKey();
		Vector<String> t2Titles = t2.getTitles();
		Vector<String> t2Keys = t2.getKey();
		String tmp = "";
		for(String s : t1Titles){
			tmp = s;
			if(t2Titles.contains(s))
				ans.add(s);
			titles.add(tmp);
			if(t1Keys.contains(s))
				keys.add(tmp);
		}
		for(String s : t2Titles){
			tmp = s;
			if(!t1Titles.contains(s))
				titles.add(tmp);
			if(t2Keys.contains(s) & !keys.contains(s))
				keys.add(tmp);
		}
		return ans;
	}
}
